package club.vann.kafka.mock.service;

import club.vann.kafka.mock.common.exception.BussinessRuntimeException;
import club.vann.kafka.mock.kafka.KafkaConsumer;
import club.vann.kafka.mock.kafka.KafkaTopicFactory;
import club.vann.kafka.mock.service.acceptor.AcceptorService;
import club.vann.kafka.mock.service.acceptor.AcceptorServiceFactory;
import com.cyberx.lunyu.AcceptorOuterClass;
import com.google.protobuf.InvalidProtocolBufferException;
import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class AcceptorInputService {
  private final AcceptorServiceFactory acceptorServiceFactory;

  public AcceptorInputService(
      AcceptorServiceFactory acceptorServiceFactory, KafkaConsumer kafkaConsumer) {
    this.acceptorServiceFactory = acceptorServiceFactory;
    kafkaConsumer.subTopic(KafkaTopicFactory.ACCEPTOR_INPUT, "", this::sub);
  }

  private void sub(byte[] bytes) {
    try {
      AcceptorOuterClass.Acceptor.AcceptorRequest acceptorRequest =
          AcceptorOuterClass.Acceptor.AcceptorRequest.parseFrom(bytes);
      log.info("接收到消息：{}", acceptorRequest);
      AcceptorOuterClass.Acceptor.RequestType requestType = acceptorRequest.getType();
      Optional<AcceptorService> optionalAcceptorService =
          acceptorServiceFactory.getAcceptorService(requestType);
      if (optionalAcceptorService.isEmpty()) {
        throw new BussinessRuntimeException(
            String.format("未识别的AcceptorRequestType：%s", requestType));
      }
      optionalAcceptorService.get().input(acceptorRequest);
    } catch (InvalidProtocolBufferException e) {
      log.error(e.getMessage());
    }
  }
}
